#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/msg.h>
#include "cJSON.h"
#include "MQTTClient.h"
#include "unistd.h"
#define ADDRESS "tcp://192.168.55.244:1883"  // MQTT代理服务器地址
#define PUB_CLIENTID "ExampleClientPub" // 发布者客户端ID
#define SUB_CLIENTID "ExampleClientSub" // 订阅者客户端ID
#define PUB_TOPIC "/app/data/up"        // 发布主题
#define SUB_TOPIC "/app/data/down"      // 订阅主题
#define QOS         1
#define TIMEOUT     10000L

struct msgbuf
{
    long mtype;
    char mtext[1024];
};

volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int report_mode;
    int msgid;
    msgid = msgget(0x001,IPC_CREAT|0777);
    if(msgid < 0)
    {
	perror("msget");
	return -1;
    }
    printf("消息队列创建成功\n");

    char *payloadptr = message->payload;    // 获取消息负载
    payloadptr[message->payloadlen] = '\0'; // 确保字符串以null结尾
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    cJSON *root = cJSON_Parse(message->payload);
    if(root)
    {
        char * json = cJSON_Print(root);
        if(json)
        {
            printf("%s\n",json);
            free (json);
        }
        cJSON_Delete(root);
    }
    
     cJSON *report_mode_json = cJSON_GetObjectItem(root, "mode");
    if (report_mode_json != NULL && report_mode_json->type == cJSON_Number)
    {
        report_mode = report_mode_json->valueint; // 获取上报模式
        printf("QT发来的上报模式: %d\n", report_mode);
    }

    int msg_id = msgget(0x001, IPC_CREAT | 0777); // 创建消息队列
    if (msg_id < 0)
    {
        perror("msgget");
        cJSON_Delete(root);
        return -1;
    }

    printf("消息队列创建成功！\n");

    // 将 JSON 对象转换为字符串
    char *json_string = cJSON_Print(root);
    if (json_string == NULL)
    {
        printf("Failed to print cJSON\n");
        cJSON_Delete(root);
        return -1;
    }

    // 计算 JSON 字符串长度
    size_t json_len = strlen(json_string);

    // 创建消息结构体
    struct msgbuf mb;
    mb.mtype = 1;
    mb.mtext[0] = '\0'; // 初始化 mtext
    strncpy(mb.mtext, json_string, sizeof(mb.mtext) - 1);
    mb.mtext[sizeof(mb.mtext) - 1] = '\0'; // 确保字符串以 null 结尾
    printf("mb.mtext:%s\n", mb.mtext);

    int ret = msgsnd(msg_id, &mb, json_len + 1, 0); // 发送消息
    if (ret < 0)
    {
        perror("msgsnd");
        free(json_string);
        cJSON_Delete(root);
        return -1;
    }

    printf("消息发送成功！\n");

    free(json_string);
    cJSON_Delete(root);    
    
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

void subscribe_to_mqtt()
{
    MQTTClient client;                                                           // 客户端句柄（描述符）
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer; // 初始化连接选项
    int rc;                                                                      // 返回值

    // 创建客户端，并且指定客户端连接的mqtt服务器地址和客户端ID
    MQTTClient_create(&client, ADDRESS, SUB_CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20; // 设置保持连接的间隔为20秒
    conn_opts.cleansession = 1;       // 设置为1表示每次连接都是一个干净的会话，不会保留之前的订阅信息

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    // 尝试连接到MQTT代理，如果连接失败则打印错误信息并退出程序
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("连接失败，返回代码 %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // 订阅指定的主题
    MQTTClient_subscribe(client, SUB_TOPIC, QOS);

    while (1)
    {
        MQTTClient_yield(); // 调用yield函数，处理等待的消息
        sleep(1);           // 休眠1秒，减少CPU使用
    }
    

    // MQTTClient_unsubscribe(client, SUB_TOPIC);
    // MQTTClient_disconnect(client, 10000);
    // MQTTClient_destroy(&client);
}
void *thread_subscribe(void *arg)
{
    subscribe_to_mqtt();
    return NULL;
}


void publish_to_mqtt(const char*message)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    MQTTClient_create(&client, ADDRESS, PUB_CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    
    char * payload = strdup(message);
    pubmsg.payload = payload;
    pubmsg.payloadlen = (int)strlen(payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), payload, PUB_TOPIC, PUB_CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);
    MQTTClient_disconnect(client,10000);
    MQTTClient_destroy(&client);
    free(payload);
}

void *thread_publish(void *arg)
{
    int shmid = *(int*)arg;
    char*json_string;
    cJSON *json_data,*temp,*hum,*lht;
    pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
    while(1)
    {
	pthread_mutex_lock(&lock);
	shmid = shmget(0x007, 1024, IPC_CREAT | 0666); // 创建共享内存，键值0x007，大小1024
        if (shmid < 0)                                       // 返回值<0共享内存创建失败
        {
            perror("shmget");
            pthread_mutex_unlock(&lock); // 释放之前获取的互斥锁，允许其他线程获取该锁
            return NULL;
        }

        json_string = shmat(shmid, NULL, 0); // 把共享内存的数据读出来
        if (json_string == (char *)-1)        // 返回值为(char *)-1就说明共享内存读取失败
        {
            perror("shmat");
            pthread_mutex_unlock(&lock);
            return NULL;
        }

        json_data = cJSON_Parse(json_string); // 解析 json_string 指向的JSON字符串
        if (!json_data)                       // 检查 json_data 是否为 NULL
        {
            printf("JSON解析失败\n");
            pthread_mutex_unlock(&lock);
            continue; // 跳出本次线程，后面的代码也不执行了，下次继续
        }
	temp = cJSON_GetObjectItem(json_data, "temp"); 
        hum = cJSON_GetObjectItem(json_data, "hum");
        lht = cJSON_GetObjectItem(json_data,"lht");
	if (!temp || !hum || !lht) // 检查 temp 或 humi 或 lht 是否为 NULL
        {
            printf("JSON中缺少温度或湿度或光照强度数据\n");
            cJSON_Delete(json_data); // 释放内存
            pthread_mutex_unlock(&lock);
            continue;
        }
	
	  // 判断上报模式
        if (report_mode == 1) // 循环上报
        {
            // 如果上报模式是1，无条件发送所有数据
            printf("上报模式1：无条件发送数据: %s\n", json_string);
            publish_to_mqtt(json_string);
            strcpy(last_data, json_string); // 更新历史数据
        }
        else if (report_mode == 2) // 刷新上报
        {
            // 如果上报模式是2，收到指令后立即发布当前最新的数据，然后停止发送
            printf("上报模式2：收到指令，发送最新数据: %s\n", json_string);
            publish_to_mqtt(json_string);
            strcpy(last_data, json_string); // 更新历史数据
            // 将上报模式设置为0或其他非发送模式，以停止进一步发送
            report_mode = 0;
        }
        else if (report_mode == 3) // 对比上报
        {
            // 如果上报模式是3，只有在数据发生变化时才发送数据
            if (strcmp(last_data, json_string) != 0)
            {
                printf("上报模式3：数据变化，发送数据: %s\n", json_string);
                publish_to_mqtt(json_string);
                strcpy(last_data, json_string); // 更新历史数据
            }
        }

	cJSON_Delete(json_data);     // 删除一个cJSON对象
        pthread_mutex_unlock(&lock); // 解互斥锁
        shmdt(json_string);
        sleep(1);
    }
    
    return NULL;
}

int main()
{
    int shmid;
    shmid = shmget(0x007,1024,IPC_CREAT|0666);
    if(shmid <0)
    {
	perror("shmget");
	return 1;
    }
    printf("共享内存创建成功\n");
    
    char *p = NULL;
    p = shmat(shmid , NULL , 0);
    if( p==(void*)-1 )
    {
        perror("shmat");
        return -1;
    }

    pthread_t tid_subscribe, tid_publish;

    pthread_create(&tid_subscribe,NULL,thread_subscribe, NULL);
    pthread_create(&tid_publish, NULL,thread_publish, &shmid);

    pthread_join(tid_subscribe, NULL);
    pthread_join(tid_publish, NULL);
    
    shmdt(p);
    shmctl(shmid, IPC_RMID, NULL);
    return 0;
}
